package com.group1.realtime.app.dws;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.app.BaseAppV1;
import com.group1.realtime.bean.TradeProvinceOrder;
import com.group1.realtime.common.Constant;
import com.group1.realtime.util.*;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;

public class Dws_TradeProvinceOrder extends BaseAppV1 {

    public static void main(String[] args) {
        new Dws_TradeProvinceOrder().init(
                3042,
                2,
                "Dws_TradeProvinceOrder",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        SingleOutputStreamOperator<TradeProvinceOrder> orderCountStream = processStream(stream);

        //添加维度
        SingleOutputStreamOperator<TradeProvinceOrder> beanStream = addDim(orderCountStream);
        //orderCountStream.print();
        SingleOutputStreamOperator<TradeProvinceOrder> aggregatedStream = windowAndAgg(beanStream);

        // 3. 把结果写入到clickHouse中
        writeToClickHouse(aggregatedStream);

    }

    private SingleOutputStreamOperator<TradeProvinceOrder> addDim(SingleOutputStreamOperator<TradeProvinceOrder> orderCountStream) {

        return orderCountStream.map(
                new RichMapFunction<TradeProvinceOrder, TradeProvinceOrder>() {
                    private Jedis redisClient;
                    private Connection conn;
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 1. 获取Phoenix连接
                        DruidDataSource dataSource = DruidDSUtil.createDataSource();
                        conn = dataSource.getConnection();

                        redisClient = RedisUtil.getRedisClient();
                    }

                    @Override
                    public void close() throws Exception {
                        // 关闭Phoenix连接
                        if (conn != null) {
                            conn.close();
                        }

                        if (redisClient != null) {
                            redisClient.close();  // 如果是手动创建的客户端是关闭.
                            // 如果是从连接池获取的客户端, 这个是归还
                        }
                    }

                    @Override
                    public TradeProvinceOrder map(TradeProvinceOrder bean) throws Exception {
                        JSONObject provinceInfo = DimUtil.readDim(redisClient, conn, "dim_province", bean.getProvinceId());

                        bean.setProvinceName(provinceInfo.getString("NAME"));
                        return bean;
                    }
                }
        );
    }

    private void writeToClickHouse(SingleOutputStreamOperator<TradeProvinceOrder> aggregatedStream) {
        aggregatedStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_trade_order_province_window", TradeProvinceOrder.class));

    }

    private SingleOutputStreamOperator<TradeProvinceOrder> windowAndAgg(SingleOutputStreamOperator<TradeProvinceOrder> orderCountStream) {

        return orderCountStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<TradeProvinceOrder>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                        .withIdleness(Duration.ofSeconds(5))
                ).keyBy(TradeProvinceOrder::getProvinceId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(
                        new ReduceFunction<TradeProvinceOrder>() {
                            @Override
                            public TradeProvinceOrder reduce(TradeProvinceOrder value1, TradeProvinceOrder value2) throws Exception {
                                value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());
                                value1.setOrderNumCt(value1.getOrderNumCt() + value2.getOrderNumCt());
                                value1.setOrderTimeCt(value1.getOrderTimeCt() + value2.getOrderTimeCt());

                                return value1;
                            }
                        },
                        new ProcessWindowFunction<TradeProvinceOrder, TradeProvinceOrder, String, TimeWindow>() {
                            @Override
                            public void process(String key, Context context, Iterable<TradeProvinceOrder> elements, Collector<TradeProvinceOrder> collector) throws Exception {
                                TradeProvinceOrder bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));
                                bean.setTs(context.currentProcessingTime());  // 把ts更新为聚合时
                                collector.collect(bean);
                            }
                        }
                );

    }

    private SingleOutputStreamOperator<TradeProvinceOrder> processStream(DataStreamSource<String> stream) {

        return stream.map(JSON::parseObject)
                .keyBy(object -> object.getString("id"))
                .process(new KeyedProcessFunction<String, JSONObject, TradeProvinceOrder>() {

                    private ValueState<String> orderState;
                    private ValueState<String> userState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        orderState = getRuntimeContext().getState(new ValueStateDescriptor<String>("orderState", String.class));
                        userState = getRuntimeContext().getState(new ValueStateDescriptor<String>("userState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject object, Context context, Collector<TradeProvinceOrder> collector) throws Exception {
                        String provinceId = object.getString("province_id");
                        Double finalAmount = object.getDouble("final_amount");
                        String orderId = object.getString("order_id");
                        String userId = object.getString("user_id");
                        Long ts = object.getLong("od_ts") * 1000;

                        if (orderState.value() == null){
                            orderState.update(orderId);
                            //System.out.println("TEST:"+object.toJSONString());
                            collector.collect(new TradeProvinceOrder(
                                    "", "",provinceId,"",finalAmount,0L,1L,ts
                            ));
                            if (userState.value() == null){
                                userState.update(userId);
                                collector.collect(new TradeProvinceOrder(
                                        "", "",provinceId,"",0.0,1L,0L,ts
                                ));
                            }
                        }

                    }
                });
    }
}
